package rsc.flow;

import java.util.*;
import java.util.concurrent.Callable;

import org.reactivestreams.*;

/**
 * A micro API for stream fusion, in particular marks producers that support a {@link QueueSubscription}.
 */
public interface Fuseable {

    /** Operational mode constant: Running with regular, arbitrary source. */
    int NONE = 0;
    /** Operational mode constant: Running with a source that acts as a synchronous source. */
    int SYNC = 1;
    /** Operational mode constant: Running with a source that acts as an asynchronous source. */
    int ASYNC = 2;
    /** Operational mode constant: QueueSubscription should decide what fusion it performs (input only). */
    int ANY = 3;
    /** 
     * Indicates that the queue will be drained from another thread 
     * thus any queue-exit computation may be invalid at that point.
     * <p>
     * For example, an {@code asyncSource.map().observeOn().subscribe()} sequence where {@code asyncSource}
     * is async-fuseable, observeOn may fuse the whole sequence into a single Queue which invokes the mapper
     * function on its {@code poll()} method that is on another thread that whereas the unfused sequence
     * would have invoked the mapper on the previous thread. If such mapper is costly, it would escape its
     * thread bound this way.
     */
    int THREAD_BARRIER = 4;

    /**
     * A subscriber variant that can immediately tell if it consumed
     * the value or not, avoiding the usual request(1) for dropped
     * values.
     *
     * @param <T> the value type
     */
    interface ConditionalSubscriber<T> extends Subscriber<T> {
        /**
         * Try consuming the value and return true if successful.
         * @param t the value to consume
         * @return true if consumed, false if dropped and a new value can be immediately sent
         */
        boolean tryOnNext(T t);
    }

    /**
     * Contract queue-fusion based optimizations for supporting subscriptions.
     *
     * <ul>
     *  <li>
     *  Synchronous sources which have fixed size and can
     *  emit its items in a pull fashion, thus avoiding the request-accounting
     *  overhead in many cases.
     *  </li>
     *  <li>
     *  Asynchronous sources which can act as a queue and subscription at
     *  the same time, saving on allocating another queue most of the time.
     * </li>
     * </ul>
     *
     * <p>
     *
     * @param <T> the value type emitted
     */
    interface QueueSubscription<T> extends Queue<T>, Subscription {

        /**
         * Request a specific fusion mode from this QueueSubscription.
         * <p>
         * One should request either SYNC, ASYNC or ANY modes (never NONE)
         * and the implementor should return NONE, SYNC or ASYNC (never ANY).
         * <p>
         * For example, if a source supports only ASYNC fusion but
         * the intermediate operator supports only SYNC fuseable sources,
         * the operator may request SYNC fusion and the source can reject it via
         * NONE, thus the operator can return NONE as well to dowstream and the
         * fusion doesn't happen.
         *
         * @param requestedMode the mode to request
         * @return the fusion mode activated
         */
        int requestFusion(int requestedMode);

        @Override
        default T peek() {
            throw new UnsupportedOperationException("Operators should not use this method!");
        }

        @Override
        default boolean add(T t) {
            throw new UnsupportedOperationException("Operators should not use this method!");
        }

        @Override
        default boolean offer(T t) {
            throw new UnsupportedOperationException("Operators should not use this method!");
        }

        @Override
        default T remove() {
            throw new UnsupportedOperationException("Operators should not use this method!");
        }

        @Override
        default T element() {
            throw new UnsupportedOperationException("Operators should not use this method!");
        }

        @Override
        default boolean contains(Object o) {
            throw new UnsupportedOperationException("Operators should not use this method!");
        }

        @Override
        default Iterator<T> iterator() {
            throw new UnsupportedOperationException("Operators should not use this method!");
        }

        @Override
        default Object[] toArray() {
            throw new UnsupportedOperationException("Operators should not use this method!");
        }

        @Override
        default <T1> T1[] toArray(T1[] a) {
            throw new UnsupportedOperationException("Operators should not use this method!");
        }

        @Override
        default boolean remove(Object o) {
            throw new UnsupportedOperationException("Operators should not use this method!");
        }

        @Override
        default boolean containsAll(Collection<?> c) {
            throw new UnsupportedOperationException("Operators should not use this method!");
        }

        @Override
        default boolean addAll(Collection<? extends T> c) {
            throw new UnsupportedOperationException("Operators should not use this method!");
        }

        @Override
        default boolean removeAll(Collection<?> c) {
            throw new UnsupportedOperationException("Operators should not use this method!");
        }

        @Override
        default boolean retainAll(Collection<?> c) {
            throw new UnsupportedOperationException("Operators should not use this method!");
        }
    }

    /**
     * Base class for synchronous sources which have fixed size and can
     * emit its items in a pull fashion, thus avoiding the request-accounting
     * overhead in many cases.
     *
     * @param <T> the content value type
     */
    interface SynchronousSubscription<T> extends QueueSubscription<T>, Queue<T> {

        @Override
        default int requestFusion(int requestedMode) {
            return Fuseable.SYNC;
        }

    }

    /**
     * Marker interface indicating that the target can return a value or null
     * immediately and thus a viable target for assembly-time optimizations.
     *
     * @param <T> the value type returned
     */
    interface ScalarCallable<T> extends Callable<T> {
        @Override
        T call();
    }
}
